package cn.kgc.gmall.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * kafka的获取连接对象的工具类
 */
public class MyKafkaUtils {

    private static final String BOOTSTRAP_SERVERS = "hadoop102:9092,hadoop103:9092,hadoop104:9092";
    private static final String DEFAULT_TOPIC = "default_topic";

    /**
     * 获取kafka消费者数据源对象
     * @return
     */
    public static FlinkKafkaConsumer<String> getKafkaConsumerSource(String topic,String groupId){
        // 6.读取kafka中的数据
        Properties prop = new Properties();
        // 连接kafka的配置
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG,groupId);

        return     new FlinkKafkaConsumer<String>(
                    topic,
                    new SimpleStringSchema(),
                    prop

                   );

    }

    /**
     * 获取kafka生产着数据源对象
     * @param topic
     * @return
     */
    public static FlinkKafkaProducer<String> getKafkaProducerSink(String topic){
        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);
        // 设置kafkfa的超时时间 默认情况下超时时间1分钟
        prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000+"");
        return    new FlinkKafkaProducer<String>(DEFAULT_TOPIC, new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
                return new ProducerRecord<>(topic,element.getBytes());
            }
        },prop, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    /**
     * 1.自定义序列化器 告诉kafka主题名是什么 如果没有名字使用默认名字
     * 2.开启kafka事务 设置精准一次提交 防止数据重复和丢失
     * @param kafkaSerializationSchema
     * @return
     */
    public static <T>FlinkKafkaProducer<T> getKafkaProducerSink(KafkaSerializationSchema<T> kafkaSerializationSchema){
        Properties prop = new Properties();
        prop.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVERS);
        // 设置kafkfa的超时时间 默认情况下超时时间1分钟
        prop.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000+"");
        // 默认主题:如果没有指定主题那么就存入此主题中  也可以从序列化器中获取主题
        return    new FlinkKafkaProducer<T>(DEFAULT_TOPIC,kafkaSerializationSchema,prop, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

    }

    //拼接Kafka相关属性到DDL
    public static String getKafkaDDL(String topic,String groupId){
        String ddl="'connector' = 'kafka', " +
                " 'topic' = '"+topic+"',"   +
                " 'properties.bootstrap.servers' = '"+ BOOTSTRAP_SERVERS +"', " +
                " 'properties.group.id' = '"+groupId+ "', " +
                "  'format' = 'json', " +
                "  'scan.startup.mode' = 'latest-offset'  ";
        return  ddl;
    }



}
